package com.sto.phoenix.appender;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.UnsynchronizedAppenderBase;

public class TransportHBaseLoggerAppender extends UnsynchronizedAppenderBase<ILoggingEvent> {

	private final Logger logger = LoggerFactory.getLogger(this.getClass());
	private Configuration configuration;
	private Connection connection;
	private Admin admin;

	/**
	 * 初始化链接
	 */
	public void init() {
		if (configuration == null) {
			configuration = HBaseConfiguration.create();
			configuration.set("hbase.zookeeper.property.clientPort", "2181");
			configuration.set("hbase.zookeeper.quorum", "192.168.122.181");
			configuration.set("hbase.master", "192.168.122.185:6000");
			configuration.set("hbase.rootdir", "hdfs://192.168.122.185:9000/hbase");
			try {
				connection = ConnectionFactory.createConnection(configuration);
				admin = connection.getAdmin();
			} catch (IOException e) {
				logger.error(e.getMessage());
				e.printStackTrace();
			}
		}
	}

	/**
	 * 插入单行
	 * 
	 * @param tableName
	 *            表名称
	 * @param rowKey
	 *            RowKey
	 * @param colFamily
	 *            列族
	 * @param col
	 *            列
	 * @param value
	 *            值
	 * @throws IOException
	 */
	public void insert(String tableName, String rowKey, String colFamily, String col, String value) throws IOException {
		init();
		Table table = connection.getTable(TableName.valueOf(tableName));
		Put put = new Put(Bytes.toBytes(rowKey));
		put.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(col), Bytes.toBytes(value));
		table.put(put);
		/*
		 * 批量插入 List<Put> putList = new ArrayList<Put>(); puts.add(put);
		 * table.put(putList);
		 */
		table.close();
	}

	@Override
	protected void append(ILoggingEvent eventObject) {
		try {
			String content = eventObject.getFormattedMessage();
			System.out.println("content内容是: " + content);
			// 插入Hbase数据库
			String tableName = "test";
			String colFamily = "test_colFamily1";
			String col = "test_col1";
			// insert(tableName, UUID.randomUUID().toString(), colFamily, col,
			// content);
		} catch (Throwable sqle) {
			System.out.println("错误");
		}

	}

}
